package k_meansClustering;

import java.io.IOException;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
           
public class K_meansClustering {
	  
	  public static void main(String[] args) throws Exception {
		  int iteration = 0, num_of_iteration = 30;
		  int feature_size =2;
		  FileSystem fs;
		  int number_of_clusters = 2;
		  
		  do {
			  Configuration conf = new Configuration();
			  fs = FileSystem.get(conf);
	           
			  Job job = new Job(conf, "K_meansClusteringMapReduce");
			  job.setJarByClass(K_meansClustering.class);
			  
			  conf = job.getConfiguration(); // This line is mandatory. 
			  
			  job.setOutputKeyClass(LongWritable.class);
			  job.setOutputValueClass(FloatArrayWritable.class);
		           
			  job.setMapperClass(K_meansClusteringMapper.class);
			  job.setReducerClass(K_meansClusteringReducer.class);
		           
			  job.setInputFormatClass(TextInputFormat.class);
			  job.setOutputFormatClass(TextOutputFormat.class);
		  
			  job.setNumReduceTasks(1); // set number of reducers to one.
			  
			  FileInputFormat.addInputPath(job, new Path(args[0]));
			  Path out = new Path(args[1]);
			  if (fs.exists(out)) fs.delete(out, true);
			  
			  FileOutputFormat.setOutputPath(job, out);
			  number_of_clusters = Integer.parseInt(args[2]);
			  num_of_iteration = Integer.parseInt(args[3]);
			  feature_size = Integer.parseInt(args[4]);
			  
			  conf.setInt("number_of_clusters", number_of_clusters);
			  conf.setInt("feature_size", feature_size);
			  conf.setInt("current_iteration_num", iteration);
			  
			  try {
				  job.waitForCompletion(true);
				  iteration++;
			  } catch (IOException e) {
				  e.printStackTrace();
			  }
		 } while (iteration < num_of_iteration);
		    
    
	  }
   }
